package com.cloud.rou.im.client.msg;

import com.cloud.rou.im.client.context.UserContextHolder;
import com.cloud.rou.im.core.util.SingletonUtil;
import com.cloud.rou.im.msg.ChatMsg;
import io.netty.channel.ChannelHandlerContext;

import java.util.Map;
import java.util.concurrent.*;

/**
 * <p></p>
 *
 * @author GaoShuangWen
 */
public class ChatMsgQueueExecutor {

    private final ExecutorService subExecutor;

    private final DelayQueue<AckDelay<ChatMsg>> delayQueue;

    private Map<Long, AckDelay<ChatMsg>> cacheMap;

    private ChatMsgQueueExecutor() {
        delayQueue = new DelayQueue<>();
        cacheMap = new ConcurrentHashMap<>();
        this.subExecutor = Executors.newFixedThreadPool(1, r -> {
            Thread thread = new Thread(r);
            thread.setDaemon(true);
            thread.setName("com.cloud.rou.im.client.msg.MessageQueueExecutor");
            return thread;
        });
        start();

    }

    public static ChatMsgQueueExecutor instance() {
        return SingletonUtil.get(ChatMsgQueueExecutor.class);
    }

    class SubWork implements Runnable {
        @Override
        public void run() {
            while (true) {
                try {
                    AckDelay<ChatMsg> ackDelay = delayQueue.take();
                    if(ackDelay.decrementAndGet()==-1){
                        System.out.println("次数已经使用完");
                    }else {
                        ChatMsg data = ackDelay.getData();
                        UserContextHolder.writeAndFlush(data);
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

        }
    }

    public void add(ChatMsg msg) {
        AckDelay<ChatMsg> e = new AckDelay<>(msg, 3000L,3);
        delayQueue.offer(e);
        cacheMap.put(msg.getMsgId(), e);
    }

    public void start() {
        subExecutor.submit(new SubWork());
    }

    public void remove(ChatMsg msg) {
        Long msgId = msg.getMsgId();
        AckDelay<ChatMsg> ackDelay = cacheMap.get(msgId);
        delayQueue.remove(ackDelay);
        cacheMap.remove(msgId);
    }
}

